﻿using BCCommon;
using BCEntity.MongoDB.PushMessage;
using BCMongoEntity;
using KafkaNet;
using KafkaNet.Model;
using KafkaNet.Protocol;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace BCService.Kafka
{

    public class KafkaProducerService : IKafkaProducerService, IDisposable
    {
        private KafkaOptions options;
        private BrokerRouter router;
        private string KafkaConnectionString { get; set; }
        private string KafkaEnvironmentType { get; set; }
        public KafkaProducerService(string kafkaConnectionString, string kafkaEnvironmentType)
        {
            this.KafkaConnectionString = kafkaConnectionString;
            this.KafkaEnvironmentType = kafkaEnvironmentType;
        }

        private int PartitionId { get; set; }
        private long Offset { get; set; }



        public void Send<T>(T entity, string kafkaTopic) where T : MongoDBBaseEntity
        {
            entity.TemplateMessageId = Guid.NewGuid().ToString().Replace("-", "");
            SendMessage(JsonConvert.SerializeObject(entity), kafkaTopic);
        }

        #region 发送消息
        private void SendMessage(string json, string kafkaTopic)
        {
            Init();
            using (Producer client = new Producer(router))
            {
                string kafkaTopicEnumStr = kafkaTopic.ToString();
                string kafkaEnvironmentStr = KafkaEnvironmentType;
                if (!string.IsNullOrWhiteSpace(kafkaEnvironmentStr))
                {
                    if (kafkaEnvironmentStr != "Production")
                    {
                        kafkaTopicEnumStr = string.Format("{0}{1}", kafkaEnvironmentStr, kafkaTopicEnumStr);
                    }
                }
                CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(5));

                var response = client.SendMessageAsync(kafkaTopicEnumStr.ToSnakeCase(), new[] { new Message(json, json) });
                response.Wait(cancellationTokenSource.Token);

                foreach (var item in response.Result)
                {
                    if (item.Error != 0)
                    {
                        throw new ArgumentException(string.Format("Kafka发送异常：\r\nTopic:{0} \r\nPartitionId:{1} \r\nOffset:{2}", item.Topic, item.PartitionId, item.Offset));
                    }
                }
            }
        }
        #endregion


        private void Init()
        {
            string url = KafkaConnectionString;
            List<string> urls = new List<string>();
            if (url.IndexOf(",") > -1)
            {
                urls = url.Split(',').ToList();
            }
            else
            {
                urls.Add(url);
            }
            List<Uri> uris = new List<Uri>();
            foreach (string item in urls)
            {
                uris.Add(new Uri(item));
            }
            options = new KafkaOptions(uris.ToArray());
            router = new BrokerRouter(options);
        }

        public void Dispose()
        {
            if (router != null)
            {
                router.Dispose();
            }
        }

        public void SendPushMessage(PushMessageMongoEntity pushMessageMongoEntity)
        {
            SendMessage(JsonConvert.SerializeObject(pushMessageMongoEntity), KafkaTopicEnum.PushMessage.ToString());
        }
    }
}
